package com.xydpeng.rpc.protocol;

import com.xydpeng.rpc.common.ApplicationConfig;
import com.xydpeng.rpc.common.InstanceMeta;
import com.xydpeng.rpc.common.RegistryConfig;
import com.xydpeng.rpc.protocol.codec.FastRpcProtocolDecoder;
import com.xydpeng.rpc.protocol.codec.FastRpcProtocolEncoder;
import com.xydpeng.rpc.protocol.handler.RpcRequestHandler;
import com.xydpeng.rpc.registry.RegistryFactory;
import com.xydpeng.rpc.registry.RegistryService;
import com.xydpeng.rpc.registry.enums.RegistryType;
import com.xydpeng.rpc.serialize.handler.DeserializeHandler;
import com.xydpeng.rpc.serialize.handler.SerializeHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author xydpeng
 * @version v1.0
 * @date: 2023/4/8
 * @description:
 */
@Slf4j
public class RpcServerBootstrap {
    private ApplicationConfig applicationConfig;

    private final Map<Class, ServiceConfig<?>> serviceConfigMap = new HashMap<>();

    private RegistryConfig registryConfig;
    private RegistryService registryService;


    private EventLoopGroup boss;
    private EventLoopGroup worker;
    private ExecutorService businessThreadpool;

    private String serverIp;
    private int serverPort;

    private int ioThreads;
    private int businessThreads;

    private volatile boolean isStarted =false;

    public RpcServerBootstrap application(ApplicationConfig applicationConfig) {
        this.applicationConfig = applicationConfig;
        return this;
    }

    public RpcServerBootstrap services(List<ServiceConfig<?>> serviceConfigs) {
        for (ServiceConfig serviceConfig : serviceConfigs) {
            this.service(serviceConfig);
        }
        return this;
    }

    public RpcServerBootstrap service(ServiceConfig<?> serviceConfig) {
        serviceConfigMap.put(serviceConfig.getClass(), serviceConfig);
        return this;
    }

    public RpcServerBootstrap adress(String host, int port) {
        this.serverIp = host;
        this.serverPort = port;
        return this;
    }

    public RpcServerBootstrap threads(int ioThreads, int businessThreads) {
        this.ioThreads = ioThreads;
        this.businessThreads = businessThreads;
        return this;
    }

    public RpcServerBootstrap registry(RegistryConfig registryConfig) {
        this.registryConfig = registryConfig;
        RegistryType type = RegistryType.valueOf(registryConfig.getRegistryType());
        this.registryService = RegistryFactory.getInstance(registryConfig.getRegistryAddr(), type);
        return this;
    }

    public void start() {
        try {
            //初始化映射器
            HandlerMapping.init(serviceConfigMap);

            ChannelFuture future = doBind();
            //注册
            doRegist();

            future.channel().closeFuture().sync();
        } catch (Exception ex) {
            log.error("RpcServer start error", ex);
        } finally {
            close();
        }
    }


    private ChannelFuture doBind() throws InterruptedException {
        if (boss == null) {
            boss = new NioEventLoopGroup(1);
        }
        if (worker == null) {
            worker = new NioEventLoopGroup(ioThreads);
        }
        if (businessThreadpool == null) {
            businessThreads = businessThreads == 0 ? Runtime.getRuntime().availableProcessors() * 2 : businessThreads;
            businessThreadpool = new ThreadPoolExecutor(businessThreads, businessThreads, 300, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(512), new ThreadPoolExecutor.AbortPolicy());
        }

        ServerBootstrap bootstrap = new ServerBootstrap();
        ChannelFuture future = bootstrap.group(boss, worker)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                        ChannelPipeline pipeline = nioSocketChannel.pipeline();
                        pipeline.addLast("fastRpcProtocolDecoder", new FastRpcProtocolDecoder());
                        pipeline.addLast("fastRpcProtocolEncoder", new FastRpcProtocolEncoder());
                        pipeline.addLast("deserializer", new DeserializeHandler());
                        pipeline.addLast("serializer", new SerializeHandler());
                        pipeline.addLast("rpcRequestHandler", new RpcRequestHandler(businessThreadpool));
                    }
                })
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .bind(serverIp, serverPort);

        future.sync();
        log.info("server started on port {}", this.serverPort);
        return future;
    }


    private RpcServerBootstrap doRegist() throws Exception {
        InstanceMeta instanceMeta = new InstanceMeta();
        instanceMeta.setApplicationName(applicationConfig.getName());
        instanceMeta.setPort(serverPort);
        instanceMeta.setIp(serverIp);
        registryService.register(instanceMeta);
        return this;
    }


    public void close() {
        if (boss != null) {
            boss.shutdownGracefully();
        }
        if (worker != null) {
            worker.shutdownGracefully();
        }
        if (businessThreadpool != null) {
            businessThreadpool.shutdown();
        }
    }


}
